package com.huoli.zfbcoupon.component;

import java.util.Collection;

import org.apache.commons.collections.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import org.springframework.kafka.listener.MessageListenerContainer;
import org.springframework.stereotype.Component;

@Component
public class KafksListenerService {

	private Logger logger = LoggerFactory.getLogger(getClass());
	@Autowired
	private KafkaListenerEndpointRegistry registry;

	public void startKafkaListener() {
		Collection<MessageListenerContainer> containers = registry.getListenerContainers();
		if (CollectionUtils.isEmpty(containers)) {
			return;
		}
		containers.forEach(c -> {
			if (!c.isRunning()) {
				logger.info("listenId={},groupId={} will be started", c.getListenerId(), c.getGroupId(), c.isRunning());
				c.start();
			}
		});
	}

	public void stopKafkaListener() {
		Collection<MessageListenerContainer> containers = registry.getListenerContainers();
		if (CollectionUtils.isEmpty(containers)) {
			return;
		}
		containers.forEach(c -> {
			if (c.isRunning()) {
				logger.info("listenId={},groupId={} will be stopped", c.getListenerId(), c.getGroupId(), c.isRunning());
				c.stop();
			}
		});
	}
}
